package com.shujia.flink.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

public class Demo2FileSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        //有界流
        //老版本写法
        //DataStream<String> studentDS = env.readTextFile("data/students.txt");

        //构建一个source
        FileSource<String> fileSource = FileSource
                //指定读取数据的格式和路径
                .forRecordStreamFormat(new TextLineInputFormat("UTF-8"), new Path("data/students"))
                //每隔10秒扫描目录下新的文件（无界流）
                .monitorContinuously(Duration.ofSeconds(10))
                .build();

        //使用file source 构建流
        DataStream<String> studentDS = env
                .fromSource(fileSource, WatermarkStrategy.noWatermarks(), "source");

        //统计班级人数
        DataStream<Tuple2<String, Integer>> kvDS = studentDS
                .map(student -> {
                    String clazz = student.split(",")[4];
                    return Tuple2.of(clazz, 1);
                }).returns(Types.TUPLE(Types.STRING, Types.INT));

        DataStream<Tuple2<String, Integer>> countDS = kvDS.keyBy(kv -> kv.f0)
                .reduce((kv1, kv2) -> Tuple2.of(kv1.f0, kv1.f1 + kv2.f1));

        countDS.print();

        env.execute();
    }
}
